package worker

import (
	"context"
	"crontab/common"
	"time"

	"go.etcd.io/etcd/api/v3/mvccpb"
	clientv3 "go.etcd.io/etcd/client/v3"
)

type jobMgr struct {
	cli *clientv3.Client
}

var (
	G_jobMgr jobMgr
	jobEvent *common.JobEvent
)

func InitJobMgr() {
	config := clientv3.Config{
		Endpoints:   G_config.EtcdEndpoints,
		DialTimeout: time.Duration(G_config.EtcdDialTimeout) * time.Millisecond,
	}
	cli, err := clientv3.New(config)
	common.DealError("ectd", err)
	timeoutCtx, cancel := context.WithTimeout(context.Background(), config.DialTimeout)
	defer cancel()
	_, err = cli.Status(timeoutCtx, config.Endpoints[0])
	common.DealError("etcd连接不上", err)
	G_jobMgr.cli = cli
}

// 监听任务变化
func (jobMgr *jobMgr) WatchJobs() error {
	// 获取/cron/jobs/下的所有任务和当前集群的revision
	getResp, err := jobMgr.cli.Get(context.TODO(), common.JOB_SAVE_DIR, clientv3.WithPrefix())
	if err != nil {
		common.DealError("etcd获取失败", err)
	}
	for _, kvPair := range getResp.Kvs {
		job, err := common.UnpackJob(kvPair.Value)
		if err != nil {
			return err
		}
		jobEvent = common.BuildJobEvent(common.JOB_EVENT_SAVE, job)
		G_scheduler.PushJobEvent(jobEvent)
	}
	go func() {
		for watchResp := range jobMgr.cli.Watch(context.TODO(), common.JOB_SAVE_DIR, clientv3.WithRev(getResp.Header.Revision+1), clientv3.WithPrefix()) {
			for _, event := range watchResp.Events {
				switch event.Type {
				case mvccpb.PUT:
					job, err := common.UnpackJob(event.Kv.Value)
					if err != nil {
						continue
					}
					jobEvent = common.BuildJobEvent(common.JOB_EVENT_SAVE, job)
				case mvccpb.DELETE:
					jobName := common.ExtractJobName(string(event.Kv.Key))
					jobEvent = common.BuildJobEvent(common.JOB_EVENT_DELETE, &common.Job{Name: jobName})
				}
				G_scheduler.PushJobEvent(jobEvent)
			}
		}
	}()
	return nil
}
